package edu.hhu.wifispark.task.sparktask;

import java.io.Serializable;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;

import edu.hhu.wifispark.entry.Customer;
import edu.hhu.wifispark.entry.Customerlive;
import edu.hhu.wifispark.util.PropUtil;
import edu.hhu.wifispark.util.SparkProUtil;

public class CustomerLiveness extends Thread implements Serializable{
	private static final long serialVersionUID = 1L;
	public List<Customer> cList;
	String taskName = null;
	public CustomerLiveness(){
		this.taskName = "customerLiveness";
	}
	
	@SuppressWarnings("resource")
	public void run(){
		SparkConf sparkConf = new SparkConf().setAppName(taskName).setMaster(SparkProUtil.MASTER);
		JavaSparkContext jsc = new JavaSparkContext(sparkConf);
		JavaRDD<Customer> customerRDD = jsc.parallelize(cList);
		
		JavaRDD<Customerlive> cliveRDD = customerRDD.map(new Function<Customer, Customerlive>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Customerlive call(Customer customer) throws Exception {
				String mac;
				long period,avetime,liveness;
				int active,freq;
				freq = customer.getBigDecimal("freq").intValue();
				avetime = customer.getBigDecimal("totaltime").longValue() / freq;
				mac = customer.getStr("mac");
				period = Long.parseLong(customer.getStr("period"));
				if(period == -1){
					//用户就来了一次，活跃值设为0
					liveness = 0;
				}
				else{
					liveness = PropUtil.PERIOD_W*period + PropUtil.AVETIME_W*avetime + PropUtil.FREQ_W*freq;
				}
				if(liveness < PropUtil.ACTIVE_LOW){
					active = 0;
				}
				else if (liveness < PropUtil.ACTIVE_MID){
					active = 1;
				}
				else if(liveness < PropUtil.ACTIVE_HIGH){
					active = 2;
				}
				else {
					active = 3;
				}
				Customerlive clive = new Customerlive();
				clive.set("mac", mac);
				clive.set("liveness", active);
				try {
					clive.save();
				} catch (Exception e) {
					clive.update();
				}
				return clive;
			}
		});
		cliveRDD.collect();
		jsc.stop();
	}
	
	public void setCList(List<Customer> icList){
		this.cList = icList;
	}
}
